#include "sqlite3_utils.h"
#include "commtraff.h"
#include "plugin.h"
#include <securec.h>
#include <stdlib.h>
#include <string.h>

// 定义将宏转换为字符串的辅助宏
#define STR(x) #x
#define TOSTR(x) STR(x)

#define SQL_DEFAULT_LEN 256

/* The index of database context state */
#define DB_STATE_QUERYING    (0x1)
#define DB_STATE_TRANSACTION (0x1 << 1)

/* The default SQL statement */
#define SQL_DROP_TABLE "DROP TABLE "
#define SQL_BEGIN_TRANSACTION "BEGIN TRANSACTION"
#define SQL_COMMIT_TRANSACTION "COMMIT TRANSACTION"
#define SQL_ROLLBACK_TRANSACTION "ROLLBACK TRANSACTION"
#define SQL_SEARCH_IF_TABLE_EXIST "SELECT * FROM sqlite_master WHERE type ='table' AND name = '%s'"

/**
 * @brief The SQL statement of SessionTraff table.
 */
#define TABLE_NAME_OF_SESSION_TRAFF "SessionTraff"
#define SQL_CREATE_SESSION_TRAFF_TABLE "CREATE TABLE IF NOT EXISTS "TABLE_NAME_OF_SESSION_TRAFF" \
    (id INTEGER PRIMARY KEY AUTOINCREMENT, \
    localSessionName TEXT, \
    peerSessionName TEXT, \
    peerUdidHash TEXT, \
    pid INTEGER, \
    time INTEGER, \
    handle INTEGER, \
    channelId INTEGER, \
    connId INTEGER, \
    sessionId INTEGER, \
    sessionType INTEGER, \
    sendBytes INTEGER, \
    recvBytes INTEGER);"
#define SQL_INSERT_SESSION_TRAFF "INSERT INTO "TABLE_NAME_OF_SESSION_TRAFF" \
    (localSessionName, peerSessionName, peerUdidHash, pid, time, handle, channelId, connId, sessionId, sessionType, sendBytes, recvBytes) \
    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
#define SQL_SEARCH_SESSION_TRAFF_BY_TIMERANGE "SELECT * FROM "TABLE_NAME_OF_SESSION_TRAFF" \
    WHERE time >= ? AND time <= ? AND id > ? ORDER BY id LIMIT "TOSTR(ONCE_RECORD_NUMS)

/**
 * @brief The SQL statement of TransportTraff table.
 */
#define TABLE_NAME_OF_TRANSPORT_TRAFF "TransportTraff"
#define SQL_CREATE_TRANSPORT_TRAFF_TABLE "CREATE TABLE IF NOT EXISTS "TABLE_NAME_OF_TRANSPORT_TRAFF" \
    (id INTEGER PRIMARY KEY AUTOINCREMENT, \
    ifName TEXT, \
    local TEXT, \
    remote TEXT, \
    pid INTEGER, \
    time INTEGER, \
    sockId INTEGER, \
    protocol INTEGER, \
    sendBytes INTEGER, \
    sendPacks INTEGER, \
    recvBytes INTEGER, \
    recvPacks INTEGER);"
#define SQL_INSERT_TRANSPORT_TRAFF "INSERT INTO "TABLE_NAME_OF_TRANSPORT_TRAFF" \
    (ifName, local, remote, pid, time, sockId, protocol, sendBytes, sendPacks, recvBytes, recvPacks) \
    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
#define SQL_SEARCH_TRANSPORT_TRAFF_BY_TIMERANGE "SELECT * FROM "TABLE_NAME_OF_TRANSPORT_TRAFF" \
    WHERE time >= ? AND time <= ? AND id > ? ORDER BY id LIMIT "TOSTR(ONCE_RECORD_NUMS)

/**
 * @brief The SQL statement of LinkTraff table.
 */
#define TABLE_NAME_OF_LINK_TRAFF "LinkTraff"
#define SQL_CREATE_LINK_TRAFF_TABLE "CREATE TABLE IF NOT EXISTS "TABLE_NAME_OF_LINK_TRAFF" \
    (id INTEGER PRIMARY KEY AUTOINCREMENT, \
    ifName TEXT, \
    time INTEGER, \
    txBytes INTEGER, \
    txPackets INTEGER, \
    rxBytes INTEGER, \
    rxPackets INTEGER, \
    rxErrPackets INTEGER, \
    txErrPackets INTEGER, \
    rxDropPackets INTEGER, \
    txDropPackets INTEGER);"
#define SQL_INSERT_LINK_TRAFF "INSERT INTO "TABLE_NAME_OF_LINK_TRAFF" \
    (ifName, time, txBytes, txPackets, rxBytes, rxPackets, rxErrPackets, txErrPackets, rxDropPackets, txDropPackets) \
    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
#define SQL_SEARCH_LINK_TRAFF_BY_TIMERANGE "SELECT * FROM "TABLE_NAME_OF_LINK_TRAFF" \
    WHERE time >= ? AND time <= ? AND id > ? ORDER BY id LIMIT "TOSTR(ONCE_RECORD_NUMS)

#define TABLE_NAME_OF_HISYSEVENT_TRAFF "HisyseventTraff"
#define SQL_CREATE_HISYSEVENT_TRAFF_TABLE "CREATE TABLE IF NOT EXISTS "TABLE_NAME_OF_HISYSEVENT_TRAFF" \
    (id INTEGER PRIMARY KEY AUTOINCREMENT, \
    domain TEXT, \
    event TEXT, \
    field TEXT, \
    value INTEGER, \
    time INTEGER);"
#define SQL_INSERT_HISYSEVENT_TRAFF "INSERT INTO "TABLE_NAME_OF_HISYSEVENT_TRAFF" \
    (domain, event, field, value, time) \
    VALUES (?, ?, ?, ?, ?)"
#define SQL_SEARCH_HISYSEVENT_TRAFF_BY_TIMERANGE "SELECT * FROM "TABLE_NAME_OF_HISYSEVENT_TRAFF" \
    WHERE time >= ? AND time <= ? AND id > ? ORDER BY id LIMIT "TOSTR(ONCE_RECORD_NUMS)

/**
 * @brief The SQL statement of RpcTraff table.
 */
#define TABLE_NAME_OF_RPC_TRAFF "RpcTraff"
#define SQL_CREATE_RPC_TRAFF_TABLE "CREATE TABLE IF NOT EXISTS "TABLE_NAME_OF_RPC_TRAFF" \
    (id INTEGER PRIMARY KEY AUTOINCREMENT, \
    proxyHandle INTEGER, \
    stubHandle INTEGER, \
    sendDataLen INTEGER, \
    recvDataLen INTEGER, \
    recvDataSeq INTEGER, \
    sendDataSeq INTEGER, \
    time INTEGER, \
    localpid INTEGER, \
    localuid INTEGER, \
    peerpid INTEGER, \
    peeruid INTEGER);"
#define SQL_INSERT_RPC_TRAFF "INSERT INTO "TABLE_NAME_OF_RPC_TRAFF" \
    (proxyHandle, stubHandle, sendDataLen, recvDataLen, recvDataSeq, sendDataSeq, time, localpid, localuid, peerpid, peeruid) \
    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
#define SQL_SEARCH_RPC_TRAFF_BY_TIMERANGE "SELECT * FROM "TABLE_NAME_OF_RPC_TRAFF" \
    WHERE time >= ? AND time <= ? AND id > ? ORDER BY id LIMIT "TOSTR(ONCE_RECORD_NUMS)

/**
 * @brief The SQL statement of WpaTraff table.
 */
#define TABLE_NAME_OF_WPA_TRAFF "WpaTraff"
#define SQL_CREATE_WPA_TRAFF_TABLE "CREATE TABLE IF NOT EXISTS "TABLE_NAME_OF_WPA_TRAFF" \
    (id INTEGER PRIMARY KEY AUTOINCREMENT, \
    ifName TEXT, \
    signal INTEGER, \
    noise INTEGER, \
    snr INTEGER, \
    txrate INTEGER, \
    rxrate INTEGER, \
    realTxrate INTEGER, \
    realRxrate INTEGER, \
    txEfficiency INTEGER, \
    frequency INTEGER, \
    txPackets INTEGER, \
    rxPackets INTEGER, \
    chload INTEGER, \
    txBytes INTEGER, \
    rxBytes INTEGER, \
    txFailed INTEGER, \
    txRetries INTEGER, \
    time INTEGER);"
#define SQL_INSERT_WPA_TRAFF "INSERT INTO "TABLE_NAME_OF_WPA_TRAFF" \
    (ifName, signal, noise, snr, txrate, rxrate, realTxrate, realRxrate, txEfficiency, frequency, txPackets, rxPackets, chload, txBytes, rxBytes, txFailed, txRetries, time) \
    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
#define SQL_SEARCH_WPA_TRAFF_BY_TIMERANGE "SELECT * FROM "TABLE_NAME_OF_WPA_TRAFF" \
    WHERE time >= ? AND time <= ? AND id > ? ORDER BY id LIMIT "TOSTR(ONCE_RECORD_NUMS)

typedef struct {
    const char *tableName;
    const char *sqlForCreate;
} SqliteManager;

typedef int32_t (*BindParaCb)(DbContext *ctx, int32_t paraNum, uint8_t *data);
// 定义回调函数指针类型
typedef int32_t (*GetTraffByTimeRangeCb)(DbContext *ctx, uint8_t *data, int32_t idx);
// 绑定参数回调函数类型定义
typedef int32_t (*BindParamsFunc)(sqlite3_stmt *stmt, CommTraffNode *node);

static int32_t BindQueryParamCb(DbContext *ctx, int32_t paraNum, uint8_t *data);

static SqliteManager g_sqliteMgr[TABLE_NAME_ID_MAX] = {
    [TABLE_SESSION_TRAFF] = {
        .tableName = TABLE_NAME_OF_SESSION_TRAFF,
        .sqlForCreate = SQL_CREATE_SESSION_TRAFF_TABLE,
    },
    [TABLE_TRANSPORT_TRAFF] = {
        .tableName = TABLE_NAME_OF_TRANSPORT_TRAFF,
        .sqlForCreate = SQL_CREATE_TRANSPORT_TRAFF_TABLE,
    },
    [TABLE_LINK_TRAFF] = {
            .tableName = TABLE_NAME_OF_LINK_TRAFF,
            .sqlForCreate = SQL_CREATE_LINK_TRAFF_TABLE,
    },
    [TABLE_HISYSEVENT_TRAFF] = {
            .tableName = TABLE_NAME_OF_HISYSEVENT_TRAFF,
            .sqlForCreate = SQL_CREATE_HISYSEVENT_TRAFF_TABLE,
    },
    [TABLE_RPC_TRAFF] = {
        .tableName = TABLE_NAME_OF_RPC_TRAFF,
        .sqlForCreate = SQL_CREATE_RPC_TRAFF_TABLE,
    },
    [TABLE_WPA_TRAFF] = {
        .tableName = TABLE_NAME_OF_WPA_TRAFF,
        .sqlForCreate = SQL_CREATE_WPA_TRAFF_TABLE,
    },
};

// 全局数据库操作互斥锁，防止多线程并发事务冲突
static pthread_mutex_t g_dbMutex = PTHREAD_MUTEX_INITIALIZER;

// 全局单例数据库上下文
static DbContext *g_singletonDbContext = NULL;
static int32_t g_dbRefCount = 0;

static int32_t ExecuteSql(DbContext *ctx, const char *sql, uint32_t len, BindParaCb cb,
                          uint8_t *data)
{
    int32_t paraNum;
    int32_t rc;

    if (ctx == NULL) {
        ERROR("ExecuteSql: database context is NULL");
        return SQLITE_ERROR;
    }
    
    if (sql == NULL || sql[0] == '\0') {
        ERROR("ExecuteSql: SQL statement is NULL or empty");
        return SQLITE_ERROR;
    }
    
    if (len == 0) {
        ERROR("ExecuteSql: SQL length is 0");
        return SQLITE_ERROR;
    }
    rc = sqlite3_prepare_v2(ctx->db, sql, len, &ctx->stmt, NULL);
    if (rc != SQLITE_OK || ctx->stmt == NULL) {
        ERROR("ExecuteSql: sqlite3_prepare_v2 failed for SQL '%.50s...': %s (error code=%d)", 
              sql, sqlite3_errmsg(ctx->db), rc);
        return sqlite3_errcode(ctx->db);
    }
    paraNum = sqlite3_bind_parameter_count(ctx->stmt);
    if (paraNum <= 0) {
        rc = sqlite3_step(ctx->stmt);
        if (rc != SQLITE_ROW && rc != SQLITE_DONE) {
            ERROR("sqlite3_step <= 0 failed, errmsg=%s", sqlite3_errmsg(ctx->db));
        }
        return rc;
    }
    if (paraNum > 0 && cb == NULL) {
        ERROR("ExecuteSql: SQL requires %d parameters but no binding callback provided", paraNum);
        (void)sqlite3_finalize(ctx->stmt);
        ctx->stmt = NULL;
        return SQLITE_ERROR;
    }
    rc = cb(ctx, paraNum, data);
    if (rc != SQLITE_OK) {
        ERROR("ExecuteSql: parameter binding callback failed with error code %d: %s", rc, sqlite3_errmsg(ctx->db));
        (void)sqlite3_finalize(ctx->stmt);
        ctx->stmt = NULL;
        return sqlite3_errcode(ctx->db);
    }
    rc = sqlite3_step(ctx->stmt);
    if (rc != SQLITE_ROW && rc != SQLITE_DONE) {
        ERROR("ExecuteSql: sqlite3_step failed with code %d: %s", rc, sqlite3_errmsg(ctx->db));
    }
    return rc;
}

static int32_t QueryData(DbContext *ctx, const char *sql, uint32_t len, BindParaCb cb, uint8_t *data)
{
    int32_t rc;

    rc = ExecuteSql(ctx, sql, len, cb, data);
    if (rc != SQLITE_ROW) {
        (void)sqlite3_finalize(ctx->stmt);
        ctx->stmt = NULL;
    } else {
        ctx->state |= DB_STATE_QUERYING;
    }
    DEBUG("QueryData done, state=%d rc=%d", ctx->state, rc);
    return rc;
}

static int32_t QueryDataNext(DbContext *ctx)
{
    int32_t rc;

    rc = sqlite3_step(ctx->stmt);
    if (rc != SQLITE_ROW) {
        ctx->state &= ~DB_STATE_QUERYING;
        (void)sqlite3_finalize(ctx->stmt);
        ctx->stmt = NULL;
    }
    return rc;
}

static bool CheckDbContextParam(const DbContext *ctx)
{
    if (ctx == NULL) {
        ERROR("CheckDbContextParam: database context is NULL");
        return false;
    }
    if (ctx->db == NULL) {
        ERROR("CheckDbContextParam: database connection is not initialized (db=NULL)");
        return false;
    }
    if (ctx->stmt != NULL) {
        ERROR("CheckDbContextParam: database context has active statement (state=%d), must be closed first", ctx->state);
        return false;
    }
    return true;
}

static bool CheckBindOrQueryParam(const DbContext *ctx)
{
    if (ctx == NULL) {
        ERROR("CheckBindOrQueryParam: database context is NULL");
        return false;
    }
    if (ctx->db == NULL) {
        ERROR("CheckBindOrQueryParam: database connection is NULL");
        return false;
    }
    if (ctx->stmt == NULL) {
        ERROR("CheckBindOrQueryParam: prepared statement is NULL");
        return false;
    }
    return true;
}

int32_t BindParaInt(DbContext *ctx, int32_t idx, int32_t value)
{
    int32_t rc;

    if (!CheckBindOrQueryParam(ctx)) {
        return SQLITE_ERROR;
    }
    
    if (idx <= 0) {
        ERROR("BindParaInt: invalid parameter index %d (must be > 0)", idx);
        return SQLITE_ERROR;
    }
    rc = sqlite3_bind_int(ctx->stmt, idx, value);
    if (rc != SQLITE_OK) {
        ERROR("BindParaInt: failed to bind integer %d at index %d: %s (error code=%d)", value, idx, sqlite3_errmsg(ctx->db), rc);
    }
    return rc;
}

static int32_t BindParaInt64(DbContext *ctx, int32_t idx, int64_t value)
{
    int32_t rc;

    if (!CheckBindOrQueryParam(ctx) || idx <= 0) {
        ERROR("invalid parameters");
        return SQLITE_ERROR;
    }
    rc = sqlite3_bind_int64(ctx->stmt, idx, value);
    if (rc != SQLITE_OK) {
        ERROR("sqlite3_bind_int64 failed: errmsg=%s", sqlite3_errmsg(ctx->db));
    }
    return rc;
}

int32_t BindParaText(DbContext *ctx, int32_t idx, const char *value, uint32_t valueLen)
{
    int32_t rc;

    if (!CheckBindOrQueryParam(ctx) || idx <= 0 || value == NULL || value[0] == '\0' || strlen(value) != valueLen) {
        ERROR("invalid parameters");
        return SQLITE_ERROR;
    }
    rc = sqlite3_bind_text(ctx->stmt, idx, value, valueLen, SQLITE_STATIC);
    if (rc != SQLITE_OK) {
        ERROR("sqlite3_bind_text failed: errmsg=%s", sqlite3_errmsg(ctx->db));
    }
    return rc;
}

int32_t BindParaDouble(DbContext *ctx, int32_t idx, double value)
{
    int32_t rc;

    if (!CheckBindOrQueryParam(ctx) || idx <= 0) {
        ERROR("invalid parameters");
        return SQLITE_ERROR;
    }
    rc = sqlite3_bind_double(ctx->stmt, idx, value);
    if (rc != SQLITE_OK) {
        ERROR("sqlite3_bind_double failed: errmsg=%s", sqlite3_errmsg(ctx->db));
    }
    return rc;
}

static int32_t GetQueryResultColCount(DbContext *ctx, int32_t *count)
{
    if (!CheckBindOrQueryParam(ctx)) {
        ERROR("invalid parameters");
        return COMMTRAFF_INVALID_PARAM;
    }
    if ((ctx->state & DB_STATE_QUERYING) == 0) {
        ERROR("the query already closed: state=%d", ctx->state);
        return COMMTRAFF_FAIL;
    }
    *count = sqlite3_column_count(ctx->stmt);
    return COMMTRAFF_SUCCESS;
}

static int32_t GetQueryResultColText(DbContext *ctx, int32_t iCol, char *text, uint32_t len)
{
    const unsigned char *result;

    if (!CheckBindOrQueryParam(ctx) || iCol < 0 || text == NULL) {
        ERROR("invalid parameters");
        return COMMTRAFF_INVALID_PARAM;
    }
    if ((ctx->state & DB_STATE_QUERYING) == 0) {
        ERROR("the query already closed: state=%d", ctx->state);
        return COMMTRAFF_FAIL;
    }
    if (sqlite3_column_type(ctx->stmt, iCol) != SQLITE_TEXT) {
        ERROR("column type not match");
        return COMMTRAFF_FAIL;
    }
    result = sqlite3_column_text(ctx->stmt, iCol);
    if (strcpy_s(text, len, (const char *)result) != EOK) {
        ERROR("strcpy_s fail");
        return COMMTRAFF_FAIL;
    }
    return COMMTRAFF_SUCCESS;
}

static int32_t GetQueryResultColInt(DbContext *ctx, int32_t iCol, int32_t *value)
{
    if (!CheckBindOrQueryParam(ctx) || iCol < 0) {
        ERROR("invalid parameters");
        return COMMTRAFF_INVALID_PARAM;
    }
    if ((ctx->state & DB_STATE_QUERYING) == 0) {
        ERROR("the query already closed: state=%d", ctx->state);
        return COMMTRAFF_FAIL;
    }
    if (sqlite3_column_type(ctx->stmt, iCol) != SQLITE_INTEGER) {
        ERROR("column type not match");
        return COMMTRAFF_FAIL;
    }
    *value = sqlite3_column_int(ctx->stmt, iCol);
    return COMMTRAFF_SUCCESS;
}

static int32_t GetQueryResultColInt64(DbContext *ctx, int32_t iCol, int64_t *value)
{
    if (!CheckBindOrQueryParam(ctx) || iCol < 0) {
        ERROR("invalid parameters");
        return COMMTRAFF_INVALID_PARAM;
    }
    if ((ctx->state & DB_STATE_QUERYING) == 0) {
        ERROR("the query already closed: state=%d", ctx->state);
        return COMMTRAFF_FAIL;
    }
    if (sqlite3_column_type(ctx->stmt, iCol) != SQLITE_INTEGER) {
        ERROR("column type not match");
        return COMMTRAFF_FAIL;
    }
    *value = sqlite3_column_int64(ctx->stmt, iCol);
    return COMMTRAFF_SUCCESS;
}

int32_t GetQueryResultColDouble(DbContext *ctx, int32_t iCol, double *value)
{
    if (!CheckBindOrQueryParam(ctx) || iCol < 0) {
        ERROR("invalid parameters");
        return COMMTRAFF_INVALID_PARAM;
    }
    if ((ctx->state & DB_STATE_QUERYING) == 0) {
        ERROR("the query already closed: state=%d", ctx->state);
        return COMMTRAFF_FAIL;
    }
    if (sqlite3_column_type(ctx->stmt, iCol) != SQLITE_FLOAT) {
        ERROR("column type not match");
        return COMMTRAFF_FAIL;
    }
    *value = sqlite3_column_double(ctx->stmt, iCol);
    return COMMTRAFF_SUCCESS;
}

static int32_t OpenTransaction(DbContext *ctx)
{
    int32_t rc;

    if (!CheckDbContextParam(ctx)) {
        ERROR("invalid parameters");
        return COMMTRAFF_INVALID_PARAM;
    }
    if ((ctx->state & DB_STATE_TRANSACTION) != 0) {
        ERROR("already open the transaction: state=%d", ctx->state);
        return COMMTRAFF_SUCCESS;
    }
    rc = ExecuteSql(ctx, SQL_BEGIN_TRANSACTION, strlen(SQL_BEGIN_TRANSACTION), NULL, NULL);
    if (rc != SQLITE_DONE) {
        ERROR("open transaction failed");
        rc = COMMTRAFF_FAIL;
    } else {
        ctx->state |= DB_STATE_TRANSACTION;
        INFO("open transaction successfully, state=%d", ctx->state);
        rc = COMMTRAFF_SUCCESS;
    }
    (void)sqlite3_finalize(ctx->stmt);
    ctx->stmt = NULL;
    return rc;
}

static int32_t CloseTransaction(DbContext *ctx, CloseTransactionType type)
{
    int32_t rc;
    const char *sql = SQL_COMMIT_TRANSACTION;

    if (!CheckDbContextParam(ctx)) {
        ERROR("invalid parameters");
        return COMMTRAFF_INVALID_PARAM;
    }
    if ((ctx->state & DB_STATE_TRANSACTION) == 0) {
        ERROR("the transaction already closed: state=%d", ctx->state);
        return COMMTRAFF_SUCCESS;
    }
    if (type == CLOSE_TRANS_ROLLBACK) {
        sql = SQL_ROLLBACK_TRANSACTION;
    }
    rc = ExecuteSql(ctx, sql, strlen(sql), NULL, NULL);
    if (rc != SQLITE_DONE) {
        ERROR("close transaction failed");
        rc = COMMTRAFF_FAIL;
    } else {
        rc = COMMTRAFF_SUCCESS;
    }
    ctx->state &= ~DB_STATE_TRANSACTION;
    (void)sqlite3_finalize(ctx->stmt);
    ctx->stmt = NULL;
    INFO("CloseTransaction, rc=%d", rc);
    return rc;
}

// 通用批量插入函数
static int32_t GenericBatchInsert(DbContext *ctx, ListNode *list, const char *sql,
                                  BindParamsFunc bindFunc, const char *dataType)
{
    int32_t rc;
    sqlite3_stmt *stmt = NULL;
    ListNode *item = NULL;
    ListNode *nextItem = NULL;
    CommTraffNode *info = NULL;
    int32_t count = 0;

    pthread_mutex_lock(&g_dbMutex);
    if (!CheckDbContextParam(ctx) || list == NULL || IsListEmpty(list) ||
        sql == NULL || bindFunc == NULL) {
        ERROR("GenericBatchInsert: invalid parameters");
        pthread_mutex_unlock(&g_dbMutex);
        return COMMTRAFF_INVALID_PARAM;
    }

    // 开启事务以提高插入性能
    rc = OpenTransaction(ctx);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("GenericBatchInsert: OpenTransaction failed");
        pthread_mutex_unlock(&g_dbMutex);
        return rc;
    }

    // 准备SQL语句
    rc = sqlite3_prepare_v2(ctx->db, sql, -1, &stmt, NULL);
    if (rc != SQLITE_OK || stmt == NULL) {
        ERROR("GenericBatchInsert: sqlite3_prepare_v2 failed, errmsg=%s", sqlite3_errmsg(ctx->db));
        (void)CloseTransaction(ctx, CLOSE_TRANS_ROLLBACK);
        pthread_mutex_unlock(&g_dbMutex);
        return COMMTRAFF_FAIL;
    }

    // 批量插入数据
    LIST_FOR_EACH_SAFE(item, nextItem, list) {
        info = LIST_ENTRY(item, CommTraffNode, node);

        // 绑定参数
        rc = bindFunc(stmt, info);
        if (rc != SQLITE_OK) {
            ERROR("GenericBatchInsert: bind parameters failed");
            goto ERROR_CLEANUP;
        }

        // 执行插入
        rc = sqlite3_step(stmt);
        if (rc != SQLITE_DONE) {
            ERROR("GenericBatchInsert: sqlite3_step failed, errmsg=%s", sqlite3_errmsg(ctx->db));
            goto ERROR_CLEANUP;
        }

        // 重置语句以供下次使用
        rc = sqlite3_reset(stmt);
        if (rc != SQLITE_OK) {
            ERROR("GenericBatchInsert: sqlite3_reset failed, errmsg=%s", sqlite3_errmsg(ctx->db));
            goto ERROR_CLEANUP;
        }
        count++;
    }

    // 释放语句
    (void)sqlite3_finalize(stmt);

    // 提交事务
    rc = CloseTransaction(ctx, CLOSE_TRANS_COMMIT);
    pthread_mutex_unlock(&g_dbMutex);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("GenericBatchInsert: CloseTransaction failed");
        return rc;
    }

    INFO("GenericBatchInsert %s success, count=%d", dataType, count);
    return COMMTRAFF_SUCCESS;

ERROR_CLEANUP:
    (void)sqlite3_finalize(stmt);
    (void)CloseTransaction(ctx, CLOSE_TRANS_ROLLBACK);
    pthread_mutex_unlock(&g_dbMutex);
    return COMMTRAFF_FAIL;
}

// 时间范围绑定回调函数
static int32_t BindQueryParamCb(DbContext *ctx, int32_t paraNum, uint8_t *data)
{
    int32_t rc;
    int32_t idx = 1;

    if (data == NULL) {
        ERROR("BindQueryParamCb get invalid data");
        return SQLITE_ERROR;
    }

    QueryParam  *queryParam = (QueryParam *)data;
    
    // 绑定开始时间
    rc = BindParaInt64(ctx, idx, queryParam->startTime);
    if (rc != SQLITE_OK) {
        ERROR("BindQueryParamCb: bind startTime failed, idx=%d, value=%"PRId64", rc=%d", idx, queryParam->startTime, rc);
        return rc;
    }
    
    // 绑定结束时间
    rc = BindParaInt64(ctx, ++idx, queryParam->endTime);
    if (rc != SQLITE_OK) {
        ERROR("BindQueryParamCb: bind endTime failed, idx=%d, value=%"PRId64", rc=%d", idx, queryParam->endTime, rc);
        return rc;
    }
    // 绑定起始记录索引
    rc = BindParaInt64(ctx, ++idx, queryParam->startRecordIndex);
    if (rc != SQLITE_OK) {
        ERROR("BindQueryParamCb: bind startRecordIndex failed, idx=%d, value=%"PRId64", rc=%d", idx, queryParam->startRecordIndex, rc);
        return rc;
    }
    return SQLITE_OK;
}

// 获取SessionTraff数据回调函数
static int32_t GetSessionTraffByTimeRangeCb(DbContext *ctx, uint8_t *data, int32_t idx)
{
    int32_t rc;
    int32_t colCount = 0;
    
    if (data == NULL) {
        ERROR("param data is null");
        return SQLITE_ERROR;
    }
    CommTraff *commTraff = (CommTraff *)((uint8_t *)data + idx * sizeof(CommTraff));
    commTraff->traffType = SESSION_LAYER_DATA;
    SessionTraff *record = &commTraff->data.sessionTraff;
    
    rc = GetQueryResultColCount(ctx, &colCount);
    if (rc != COMMTRAFF_SUCCESS || colCount < 13) { // 增加了id字段，所以需要13列而不是12列
        ERROR("GetQueryResultColCount failed or column count not match");
        return COMMTRAFF_FAIL;
    }
    
    char tempStr[256] = {0};

    // 获取localSessionName (索引从0改为1，因为第0列是id)
    rc = GetQueryResultColText(ctx, 1, tempStr, sizeof(tempStr));
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get localSessionName failed");
        return COMMTRAFF_FAIL;
    }
    (void)strncpy_s((char *)record->localSessionName, sizeof(record->localSessionName), tempStr, sizeof(record->localSessionName) - 1);

    // 获取peerSessionName (索引从1改为2)
    rc = GetQueryResultColText(ctx, 2, tempStr, sizeof(tempStr));
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get peerSessionName failed");
        return COMMTRAFF_FAIL;
    }
    (void)strncpy_s((char *)record->peerSessionName, sizeof(record->peerSessionName), tempStr, sizeof(record->peerSessionName) - 1);

    // 获取peerUdidHash (索引从2改为3)
    rc = GetQueryResultColText(ctx, 3, tempStr, sizeof(tempStr));
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get peerUdidHash failed");
        return COMMTRAFF_FAIL;
    }
    (void)strncpy_s((char *)record->peerUdidHash, sizeof(record->peerUdidHash), tempStr, sizeof(record->peerUdidHash) - 1);

    // 获取pid (索引从3改为4)
    rc = GetQueryResultColInt(ctx, 4, (int32_t *)&record->pid);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get pid failed");
        return COMMTRAFF_FAIL;
    }

    // 获取time (索引从4改为5)
    rc = GetQueryResultColInt64(ctx, 5, (int64_t *)&record->time);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get time failed");
        return COMMTRAFF_FAIL;
    }

    // 获取handle (索引从5改为6)
    rc = GetQueryResultColInt(ctx, 6, &record->handle);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get handle failed");
        return COMMTRAFF_FAIL;
    }

    // 获取channelId (索引从6改为7)
    rc = GetQueryResultColInt(ctx, 7, (int32_t *)&record->channelId);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get channelId failed");
        return COMMTRAFF_FAIL;
    }

    // 获取connId (索引从7改为8)
    rc = GetQueryResultColInt(ctx, 8, (int32_t *)&record->connId);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get connId failed");
        return COMMTRAFF_FAIL;
    }

    // 获取sessionId (索引从8改为9)
    rc = GetQueryResultColInt(ctx, 9, (int32_t *)&record->sessionId);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get sessionId failed");
        return COMMTRAFF_FAIL;
    }

    // 获取sessionType (索引从9改为10)
    rc = GetQueryResultColInt(ctx, 10, (int32_t *)&record->sessionType);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get sessionType failed");
        return COMMTRAFF_FAIL;
    }

    // 获取sendBytes (索引从10改为11)
    rc = GetQueryResultColInt(ctx, 11, (int32_t *)&record->sendBytes);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get sendBytes failed");
        return COMMTRAFF_FAIL;
    }

    // 获取recvBytes (索引从11改为12)
    rc = GetQueryResultColInt(ctx, 12, (int32_t *)&record->recvBytes);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get recvBytes failed");
        return COMMTRAFF_FAIL;
    }

    return COMMTRAFF_SUCCESS;
}

// 获取TransportTraff数据回调函数
static int32_t GetTransportTraffByTimeRangeCb(DbContext *ctx, uint8_t *data, int32_t idx)
{
    int32_t rc;
    int32_t colCount = 0;

    if (data == NULL) {
        ERROR("param data is null");
        return SQLITE_ERROR;
    }

    CommTraff *commTraff = (CommTraff *)((uint8_t *)data + idx * sizeof(CommTraff));
    commTraff->traffType = TRANSPORT_LAYER_DATA;
    TransportTraff *record = &commTraff->data.transportTraff;

    rc = GetQueryResultColCount(ctx, &colCount);
    if (rc != COMMTRAFF_SUCCESS || colCount < 12) { // 增加了id字段，所以需要12列而不是11列
        ERROR("GetQueryResultColCount failed or column count not match");
        return COMMTRAFF_FAIL;
    }

    char tempStr[256] = {0};

    // 获取ifName (索引从0改为1，因为第0列是id)
    rc = GetQueryResultColText(ctx, 1, tempStr, sizeof(tempStr));
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get ifName failed");
        return COMMTRAFF_FAIL;
    }
    (void)strncpy_s((char *)record->ifName, sizeof(record->ifName), tempStr, sizeof(record->ifName) - 1);

    // 获取local (索引从1改为2)
    rc = GetQueryResultColText(ctx, 2, record->local, sizeof(record->local));
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get local failed");
        return COMMTRAFF_FAIL;
    }

    // 获取remote (索引从2改为3)
    rc = GetQueryResultColText(ctx, 3, record->remote, sizeof(record->remote));
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get remote failed");
        return COMMTRAFF_FAIL;
    }

    // 获取pid (索引从3改为4)
    rc = GetQueryResultColInt(ctx, 4, (int32_t *)&record->pid);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get pid failed");
        return COMMTRAFF_FAIL;
    }

    // 获取time (索引从4改为5)
    rc = GetQueryResultColInt64(ctx, 5, (int64_t *)&record->time);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get time failed");
        return COMMTRAFF_FAIL;
    }

    // 获取sockId (索引从5改为6)
    rc = GetQueryResultColInt(ctx, 6, (int32_t *)&record->sockId);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get sockId failed");
        return COMMTRAFF_FAIL;
    }

    // 获取protocol (索引从6改为7)
    rc = GetQueryResultColInt(ctx, 7, (int32_t *)&record->protocol);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get protocol failed");
        return COMMTRAFF_FAIL;
    }

    // 获取sendBytes (索引从7改为8)
    rc = GetQueryResultColInt(ctx, 8, (int32_t *)&record->sendBytes);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get sendBytes failed");
        return COMMTRAFF_FAIL;
    }

    // 获取sendPacks (索引从8改为9)
    rc = GetQueryResultColInt(ctx, 9, (int32_t *)&record->sendPacks);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get sendPacks failed");
        return COMMTRAFF_FAIL;
    }

    // 获取recvBytes (索引从9改为10)
    rc = GetQueryResultColInt(ctx, 10, (int32_t *)&record->recvBytes);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get recvBytes failed");
        return COMMTRAFF_FAIL;
    }

    // 获取recvPacks (索引从10改为11)
    rc = GetQueryResultColInt(ctx, 11, (int32_t *)&record->recvPacks);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get recvPacks failed");
        return COMMTRAFF_FAIL;
    }

    return COMMTRAFF_SUCCESS;
}

// 获取LinkTraff数据回调函数
static int32_t GetLinkTraffByTimeRangeCb(DbContext *ctx, uint8_t *data, int32_t idx)
{
    int32_t rc;
    int32_t colCount = 0;

    if (data == NULL) {
        ERROR("param data is null");
        return SQLITE_ERROR;
    }

    CommTraff *commTraff = (CommTraff *)((uint8_t *)data + idx * sizeof(CommTraff));
    commTraff->traffType = LINK_LAYER_DATA;
    LinkTraff *record = &commTraff->data.linkTraff;

    rc = GetQueryResultColCount(ctx, &colCount);
    if (rc != COMMTRAFF_SUCCESS || colCount < 11) { // 增加了id字段，所以需要11列而不是10列
        ERROR("GetQueryResultColCount failed or column count not match");
        return COMMTRAFF_FAIL;
    }

    char tempStr[256] = {0};

    // 获取ifName (索引从0改为1，因为第0列是id)
    rc = GetQueryResultColText(ctx, 1, tempStr, sizeof(tempStr));
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get ifName failed");
        return COMMTRAFF_FAIL;
    }
    (void)strncpy_s((char *)record->ifName, sizeof(record->ifName), tempStr, sizeof(record->ifName) - 1);

    // 获取time (索引从1改为2)
    rc = GetQueryResultColInt64(ctx, 2, (int64_t *)&record->time);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get time failed");
        return COMMTRAFF_FAIL;
    }

    // 获取txBytes (索引从2改为3)
    rc = GetQueryResultColInt(ctx, 3, (int32_t *)&record->txBytes);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get txBytes failed");
        return COMMTRAFF_FAIL;
    }

    // 获取txPackets (索引从3改为4)
    rc = GetQueryResultColInt(ctx, 4, (int32_t *)&record->txPackets);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get txPackets failed");
        return COMMTRAFF_FAIL;
    }

    // 获取rxBytes (索引从4改为5)
    rc = GetQueryResultColInt(ctx, 5, (int32_t *)&record->rxBytes);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get rxBytes failed");
        return COMMTRAFF_FAIL;
    }

    // 获取rxPackets (索引从5改为6)
    rc = GetQueryResultColInt(ctx, 6, (int32_t *)&record->rxPackets);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get rxPackets failed");
        return COMMTRAFF_FAIL;
    }

    // 获取rxErrPackets (索引从6改为7)
    rc = GetQueryResultColInt(ctx, 7, (int32_t *)&record->rxErrPackets);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get rxErrPackets failed");
        return COMMTRAFF_FAIL;
    }

    // 获取txErrPackets (索引从7改为8)
    rc = GetQueryResultColInt(ctx, 8, (int32_t *)&record->txErrPackets);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get txErrPackets failed");
        return COMMTRAFF_FAIL;
    }

    // 获取rxDropPackets (索引从8改为9)
    rc = GetQueryResultColInt(ctx, 9, (int32_t *)&record->rxDropPackets);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get rxDropPackets failed");
        return COMMTRAFF_FAIL;
    }

    // 获取txDropPackets (索引从9改为10)
    rc = GetQueryResultColInt(ctx, 10, (int32_t *)&record->txDropPackets);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get txDropPackets failed");
        return COMMTRAFF_FAIL;
    }

    return COMMTRAFF_SUCCESS;
}

// 获取RpcTraff数据回调函数
static int32_t GetRpcTraffByTimeRangeCb(DbContext *ctx, uint8_t *data, int32_t idx)
{
    int32_t rc;
    int32_t colCount = 0;

    if (data == NULL) {
        ERROR("param data is null");
        return SQLITE_ERROR;
    }

    CommTraff *commTraff = (CommTraff *)((uint8_t *)data + idx * sizeof(CommTraff));
    commTraff->traffType = RPC_LAYER_DATA;
    RpcTraff *record = &commTraff->data.rpcTraff;

    rc = GetQueryResultColCount(ctx, &colCount);
    if (rc != COMMTRAFF_SUCCESS || colCount < 12) { // 增加了id字段和4个新字段，所以需要12列而不是8列
        ERROR("GetQueryResultColCount failed or column count not match");
        return COMMTRAFF_FAIL;
    }

    // 获取proxyHandle (索引从0改为1，因为第0列是id)
    rc = GetQueryResultColInt(ctx, 1, &record->proxyHandle);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get proxyHandle failed");
        return COMMTRAFF_FAIL;
    }

    // 获取stubHandle (索引从1改为2)
    rc = GetQueryResultColInt(ctx, 2, &record->stubHandle);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get stubHandle failed");
        return COMMTRAFF_FAIL;
    }

    // 获取sendDataLen (索引从2改为3)
    rc = GetQueryResultColInt(ctx, 3, (int32_t *)&record->sendDataLen);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get sendDataLen failed");
        return COMMTRAFF_FAIL;
    }

    // 获取recvDataLen (索引从3改为4)
    rc = GetQueryResultColInt(ctx, 4, (int32_t *)&record->recvDataLen);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get recvDataLen failed");
        return COMMTRAFF_FAIL;
    }

    // 获取recvDataSeq (索引从4改为5)
    rc = GetQueryResultColInt64(ctx, 5, (int64_t *)&record->recvDataSeq);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get recvDataSeq failed");
        return COMMTRAFF_FAIL;
    }

    // 获取sendDataSeq (索引从5改为6)
    rc = GetQueryResultColInt64(ctx, 6, (int64_t *)&record->sendDataSeq);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get sendDataSeq failed");
        return COMMTRAFF_FAIL;
    }

    // 获取time (索引从6改为7)
    rc = GetQueryResultColInt64(ctx, 7, (int64_t *)&record->time);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get time failed");
        return COMMTRAFF_FAIL;
    }

    // 获取localpid (索引8)
    rc = GetQueryResultColInt(ctx, 8, (int32_t *)&record->localpid);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get localpid failed");
        return COMMTRAFF_FAIL;
    }

    // 获取localuid (索引9)
    rc = GetQueryResultColInt(ctx, 9, (int32_t *)&record->localuid);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get localuid failed");
        return COMMTRAFF_FAIL;
    }

    // 获取peerpid (索引10)
    rc = GetQueryResultColInt(ctx, 10, (int32_t *)&record->peerpid);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get peerpid failed");
        return COMMTRAFF_FAIL;
    }

    // 获取peeruid (索引11)
    rc = GetQueryResultColInt(ctx, 11, (int32_t *)&record->peeruid);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get peeruid failed");
        return COMMTRAFF_FAIL;
    }

    return COMMTRAFF_SUCCESS;
}

// 获取WpaTraff数据回调函数
static int32_t GetWpaTraffByTimeRangeCb(DbContext *ctx, uint8_t *data, int32_t idx)
{
    int32_t rc;
    int32_t colCount = 0;

    if (data == NULL) {
        ERROR("param data is null");
        return SQLITE_ERROR;
    }

    CommTraff *commTraff = (CommTraff *)((uint8_t *)data + idx * sizeof(CommTraff));
    commTraff->traffType = WPA_LAYER_DATA;
    WpaTraff *record = &commTraff->data.wpaTraff;

    rc = GetQueryResultColCount(ctx, &colCount);
    if (rc != COMMTRAFF_SUCCESS || colCount < 19) { // 增加了id字段，所以需要19列而不是18列
        ERROR("GetQueryResultColCount failed or column count not match");
        return COMMTRAFF_FAIL;
    }

    char tempStr[256] = {0};
    
    // 获取ifName (索引从0改为1，因为第0列是id)
    rc = GetQueryResultColText(ctx, 1, tempStr, sizeof(tempStr));
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get ifName failed");
        return COMMTRAFF_FAIL;
    }
    (void)strncpy_s((char *)record->ifName, sizeof(record->ifName), tempStr, sizeof(record->ifName) - 1);
    
    // 获取signal (索引从1改为2)
    rc = GetQueryResultColInt(ctx, 2, (int32_t *)&record->signal);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get signal failed");
        return COMMTRAFF_FAIL;
    }
    
    // 获取noise (索引从2改为3)
    rc = GetQueryResultColInt(ctx, 3, (int32_t *)&record->noise);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get noise failed");
        return COMMTRAFF_FAIL;
    }
    
    // 获取snr (索引从3改为4)
    rc = GetQueryResultColInt(ctx, 4, (int32_t *)&record->snr);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get snr failed");
        return COMMTRAFF_FAIL;
    }
    
    // 获取txrate (索引从4改为5)
    rc = GetQueryResultColInt(ctx, 5, (int32_t *)&record->txrate);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get txrate failed");
        return COMMTRAFF_FAIL;
    }
    
    // 获取rxrate (索引从5改为6)
    rc = GetQueryResultColInt(ctx, 6, (int32_t *)&record->rxrate);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get rxrate failed");
        return COMMTRAFF_FAIL;
    }
    
    // 获取realTxrate (索引从6改为7)
    rc = GetQueryResultColInt(ctx, 7, (int32_t *)&record->realTxrate);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get realTxrate failed");
        return COMMTRAFF_FAIL;
    }
    
    // 获取realRxrate (索引从7改为8)
    rc = GetQueryResultColInt(ctx, 8, (int32_t *)&record->realRxrate);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get realRxrate failed");
        return COMMTRAFF_FAIL;
    }
    
    // 获取txEfficiency (索引从8改为9)
    rc = GetQueryResultColInt(ctx, 9, (int32_t *)&record->txEfficiency);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get txEfficiency failed");
        return COMMTRAFF_FAIL;
    }
    
    // 获取frequency (索引从9改为10)
    rc = GetQueryResultColInt(ctx, 10, (int32_t *)&record->frequency);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get frequency failed");
        return COMMTRAFF_FAIL;
    }
    
    // 获取txPackets (索引从10改为11)
    rc = GetQueryResultColInt(ctx, 11, (int32_t *)&record->txPackets);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get txPackets failed");
        return COMMTRAFF_FAIL;
    }
    
    // 获取rxPackets (索引从11改为12)
    rc = GetQueryResultColInt(ctx, 12, (int32_t *)&record->rxPackets);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get rxPackets failed");
        return COMMTRAFF_FAIL;
    }
    
    // 获取chload (索引从12改为13)
    rc = GetQueryResultColInt(ctx, 13, (int32_t *)&record->chload);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get chload failed");
        return COMMTRAFF_FAIL;
    }
    
    // 获取txBytes (索引从13改为14)
    rc = GetQueryResultColInt64(ctx, 14, (int64_t *)&record->txBytes);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get txBytes failed");
        return COMMTRAFF_FAIL;
    }
    
    // 获取rxBytes (索引从14改为15)
    rc = GetQueryResultColInt64(ctx, 15, (int64_t *)&record->rxBytes);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get rxBytes failed");
        return COMMTRAFF_FAIL;
    }
    
    // 获取txFailed (索引从15改为16)
    rc = GetQueryResultColInt(ctx, 16, (int32_t *)&record->txFailed);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get txFailed failed");
        return COMMTRAFF_FAIL;
    }
    
    // 获取txRetries (索引从16改为17)
    rc = GetQueryResultColInt(ctx, 17, (int32_t *)&record->txRetries);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get txRetries failed");
        return COMMTRAFF_FAIL;
    }
    
    // 获取time (索引从17改为18)
    rc = GetQueryResultColInt64(ctx, 18, (int64_t *)&record->time);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get time failed");
        return COMMTRAFF_FAIL;
    }
    
    return COMMTRAFF_SUCCESS;
}

// 获取HisyseventTraff数据回调函数
static int32_t GetHisyseventTraffByTimeRangeCb(DbContext *ctx, uint8_t *data, int32_t idx)
{
    int32_t rc;
    int32_t colCount = 0;

    if (data == NULL) {
        ERROR("param data is null");
        return SQLITE_ERROR;
    }

    CommTraff *commTraff = (CommTraff *)((uint8_t *)data + idx * sizeof(CommTraff));
    commTraff->traffType = HISYSEVENT_DATA;
    HisyseventData *record = &commTraff->data.hisyseventData;

    rc = GetQueryResultColCount(ctx, &colCount);
    if (rc != COMMTRAFF_SUCCESS || colCount < 6) { // 增加了id字段，所以需要6列而不是5列
        ERROR("GetQueryResultColCount failed or column count not match");
        return COMMTRAFF_FAIL;
    }

    char tempStr[256] = {0};

    // 获取domain (索引从0改为1，因为第0列是id)
    rc = GetQueryResultColText(ctx, 1, tempStr, sizeof(tempStr));
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get domain failed");
        return COMMTRAFF_FAIL;
    }
    (void)strncpy_s((char *)record->domain, sizeof(record->domain), tempStr, sizeof(record->domain) - 1);

    // 获取event (索引从1改为2)
    rc = GetQueryResultColText(ctx, 2, tempStr, sizeof(tempStr));
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get event failed");
        return COMMTRAFF_FAIL;
    }
    (void)strncpy_s((char *)record->event, sizeof(record->event), tempStr, sizeof(record->event) - 1);

    // 获取field (索引从2改为3)
    rc = GetQueryResultColText(ctx, 3, tempStr, sizeof(tempStr));
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get field failed");
        return COMMTRAFF_FAIL;
    }
    (void)strncpy_s((char *)record->field, sizeof(record->field), tempStr, sizeof(record->field) - 1);

    // 获取value (索引从3改为4)
    rc = GetQueryResultColInt64(ctx, 4, (int64_t *)&record->value);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get value failed");
        return COMMTRAFF_FAIL;
    }

    // 获取time (索引从4改为5)
    rc = GetQueryResultColInt64(ctx, 5, (int64_t *)&record->time);
    if (rc != COMMTRAFF_SUCCESS) {
        ERROR("Get time failed");
        return COMMTRAFF_FAIL;
    }

    return COMMTRAFF_SUCCESS;
}

int32_t OpenDatabase(DbContext **ctx)
{
    int32_t rc;
    sqlite3 *sqlite = NULL;

    if (ctx == NULL) {
        ERROR("OpenDatabase: output context pointer is NULL");
        return COMMTRAFF_INVALID_PARAM;
    }

    pthread_mutex_lock(&g_dbMutex);
    
    // 如果单例数据库上下文已存在，直接返回并增加引用计数
    if (g_singletonDbContext != NULL) {
        *ctx = g_singletonDbContext;
        g_dbRefCount++;
        INFO("OpenDatabase: reuse existing database context, refCount=%d", g_dbRefCount);
        pthread_mutex_unlock(&g_dbMutex);
        return COMMTRAFF_SUCCESS;
    }

    // 创建新的数据库连接
    rc = sqlite3_open_v2(DATABASE_NAME, &sqlite, SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE |
                                                 SQLITE_OPEN_NOMUTEX, NULL);
    if (rc != SQLITE_OK || sqlite == NULL) {
        ERROR("OpenDatabase: failed to open database '%s': %s (sqlite error=%d, errno=%d-%s)", 
              DATABASE_NAME, sqlite ? sqlite3_errmsg(sqlite) : "unknown error", rc, errno, strerror(errno));
        if (sqlite) {
            (void)sqlite3_close_v2(sqlite);
        }
        pthread_mutex_unlock(&g_dbMutex);
        return COMMTRAFF_FAIL;
    }

    // 创建数据库上下文
    g_singletonDbContext = (DbContext *)calloc(1, sizeof(DbContext));
    if (g_singletonDbContext == NULL) {
        ERROR("OpenDatabase: failed to allocate %zu bytes for DbContext: %s (errno=%d)", sizeof(DbContext), strerror(errno), errno);
        (void)sqlite3_close_v2(sqlite);
        pthread_mutex_unlock(&g_dbMutex);
        return COMMTRAFF_MEM_ALLOC_FAIL;
    }
    
    g_singletonDbContext->db = sqlite;
    g_dbRefCount = 1;
    *ctx = g_singletonDbContext;
    
    INFO("OpenDatabase: created new database context, refCount=%d", g_dbRefCount);
    pthread_mutex_unlock(&g_dbMutex);
    return COMMTRAFF_SUCCESS;
}

int32_t CloseDatabase(DbContext *ctx)
{
    if (ctx == NULL) {
        ERROR("CloseDatabase: database context is NULL");
        return COMMTRAFF_INVALID_PARAM;
    }

    pthread_mutex_lock(&g_dbMutex);
    
    // 检查是否是单例数据库上下文
    if (ctx != g_singletonDbContext) {
        ERROR("CloseDatabase: ctx is not singleton database context");
        pthread_mutex_unlock(&g_dbMutex);
        return COMMTRAFF_INVALID_PARAM;
    }
    
    // 减少引用计数，防止变为负数
    if (g_dbRefCount > 0) {
        g_dbRefCount--;
    }
    INFO("CloseDatabase: decreased refCount to %d", g_dbRefCount);
    
    // 只有当引用计数为0时才真正关闭数据库
    if (g_dbRefCount == 0) {
        if (g_singletonDbContext->db != NULL) {
            (void)sqlite3_close_v2(g_singletonDbContext->db);
        }
        free(g_singletonDbContext);
        g_singletonDbContext = NULL;
        g_dbRefCount = 0;
        INFO("CloseDatabase: database context closed and freed");
    }
    
    pthread_mutex_unlock(&g_dbMutex);
    return COMMTRAFF_SUCCESS;
}

int32_t CreateTable(DbContext *ctx, TableNameID id)
{
    int32_t rc;
    char *errMsg = NULL;

    if (!CheckDbContextParam(ctx)) {
        ERROR("invalid parameters");
        return COMMTRAFF_INVALID_PARAM;
    }
    const char *sql = g_sqliteMgr[id].sqlForCreate;
    if (sql == NULL || sql[0] == '\0') {
        ERROR("createsql is not impl");
        return COMMTRAFF_FAIL;
    }
    rc = sqlite3_exec(ctx->db, sql, NULL, NULL, &errMsg);
    if (rc != SQLITE_OK && errMsg != NULL) {
        ERROR("sqlite_exec fail: errmsg=%s", errMsg);
        sqlite3_free(errMsg);
    }
    return rc == SQLITE_OK ? COMMTRAFF_SUCCESS : COMMTRAFF_FAIL;
}

int32_t DeleteTable(DbContext *ctx, TableNameID id)
{
    int32_t rc;
    char sql[SQL_DEFAULT_LEN] = {0};

    if (!CheckDbContextParam(ctx)) {
        ERROR("invalid parameters");
        return COMMTRAFF_INVALID_PARAM;
    }
    rc = sprintf_s(sql, SQL_DEFAULT_LEN, "%s%s", SQL_DROP_TABLE, g_sqliteMgr[id].tableName);
    if (rc < 0) {
        ERROR("sprintf_s sql fail");
        return COMMTRAFF_FAIL;
    }
    rc = ExecuteSql(ctx, sql, strlen(sql), NULL, NULL);
    if (rc != SQLITE_DONE) {
        ERROR("delete table fail");
        rc = COMMTRAFF_FAIL;
    } else {
        rc = COMMTRAFF_SUCCESS;
    }
    (void)sqlite3_finalize(ctx->stmt);
    ctx->stmt = NULL;
    return rc;
}

int32_t CheckTableExist(DbContext *ctx, TableNameID id, bool *isExist)
{
    int32_t rc;
    char sql[SQL_DEFAULT_LEN] = {0};

    if (!CheckDbContextParam(ctx) || isExist == NULL) {
        ERROR("invalid parameters");
        return COMMTRAFF_INVALID_PARAM;
    }
    rc = sprintf_s(sql, SQL_DEFAULT_LEN, SQL_SEARCH_IF_TABLE_EXIST, g_sqliteMgr[id].tableName);
    if (rc < 0) {
        ERROR("sprintf_s sql fail");
        return COMMTRAFF_FAIL;
    }
    *isExist = false;
    rc = ExecuteSql(ctx, sql, strlen(sql), NULL, NULL);
    if (rc == SQLITE_ROW && sqlite3_column_count(ctx->stmt) != 0) {
        *isExist = true;
    }
    (void)sqlite3_finalize(ctx->stmt);
    ctx->stmt = NULL;
    return COMMTRAFF_SUCCESS;
}

int32_t DeleteOlderThan(DbContext *ctx, TableNameID id, int64_t beforeTime)
{
    if (id < 0 || id >= TABLE_NAME_ID_MAX) {
        ERROR("DeleteOlderThan: invalid table id");
        return COMMTRAFF_INVALID_PARAM;
    }

    const char *tableName = g_sqliteMgr[id].tableName;
    if (tableName == NULL || tableName[0] == '\0') {
        ERROR("DeleteOlderThan: empty table name");
        return COMMTRAFF_INVALID_PARAM;
    }

    char sql[SQL_DEFAULT_LEN] = {0};
    int rc = sprintf_s(sql, SQL_DEFAULT_LEN, "DELETE FROM %s WHERE time < ?", tableName);
    if (rc < 0) {
        ERROR("DeleteOlderThan: build sql fail");
        return COMMTRAFF_FAIL;
    }

    pthread_mutex_lock(&g_dbMutex);
    if (ctx == NULL || ctx->db == NULL) {
        pthread_mutex_unlock(&g_dbMutex);
        ERROR("DeleteOlderThan: invalid ctx");
        return COMMTRAFF_INVALID_PARAM;
    }
    sqlite3_stmt *stmt = NULL;
    rc = sqlite3_prepare_v2(ctx->db, sql, -1, &stmt, NULL);
    if (rc != SQLITE_OK || stmt == NULL) {
        ERROR("DeleteOlderThan: prepare fail: %s", sqlite3_errmsg(ctx->db));
        pthread_mutex_unlock(&g_dbMutex);
        return COMMTRAFF_FAIL;
    }
    rc = sqlite3_bind_int64(stmt, 1, (sqlite3_int64)beforeTime);
    if (rc != SQLITE_OK) {
        ERROR("DeleteOlderThan: bind fail: %d", rc);
        sqlite3_finalize(stmt);
        pthread_mutex_unlock(&g_dbMutex);
        return COMMTRAFF_FAIL;
    }
    rc = sqlite3_step(stmt);
    if (rc != SQLITE_DONE) {
        ERROR("DeleteOlderThan: step fail: %d", rc);
        sqlite3_finalize(stmt);
        pthread_mutex_unlock(&g_dbMutex);
        return COMMTRAFF_FAIL;
    }
    
    // 获取删除的记录数并打印
    int deletedRows = sqlite3_changes(ctx->db);
    INFO("DeleteOlderThan: deleted %d records from %s table", deletedRows, tableName);
    
    sqlite3_finalize(stmt);
    pthread_mutex_unlock(&g_dbMutex);
    return COMMTRAFF_SUCCESS;
}

// SessionTraff绑定参数函数
static int32_t BindSessionTraffParams(sqlite3_stmt *stmt, CommTraffNode *node)
{
    int32_t rc;
    int32_t idx = 1;
    SessionTraff *record = &node->data.data.sessionTraff;
    
    rc = sqlite3_bind_text(stmt, idx++, (const char *)record->localSessionName, 
                          strlen((const char *)record->localSessionName), SQLITE_STATIC);
    if (rc != SQLITE_OK) {
        ERROR("BindSessionTraffParams: bind localSessionName failed, idx=%d, value=%s, rc=%d", idx-1, record->localSessionName, rc);
        return rc;
    }
    
    rc = sqlite3_bind_text(stmt, idx++, (const char *)record->peerSessionName, 
                          strlen((const char *)record->peerSessionName), SQLITE_STATIC);
    if (rc != SQLITE_OK) {
        ERROR("BindSessionTraffParams: bind peerSessionName failed, idx=%d, value=%s, rc=%d", idx-1, record->peerSessionName, rc);
        return rc;
    }
    
    rc = sqlite3_bind_text(stmt, idx++, (const char *)record->peerUdidHash, 
                          strlen((const char *)record->peerUdidHash), SQLITE_STATIC);
    if (rc != SQLITE_OK) {
        ERROR("BindSessionTraffParams: bind peerUdidHash failed, idx=%d, value=%s, rc=%d", idx-1, record->peerUdidHash, rc);
        return rc;
    }
    
    rc = sqlite3_bind_int(stmt, idx++, record->pid);
    if (rc != SQLITE_OK) {
        ERROR("BindSessionTraffParams: bind pid failed, idx=%d, value=%d, rc=%d", idx-1, record->pid, rc);
        return rc;
    }
    
    rc = sqlite3_bind_int64(stmt, idx++, record->time);
    if (rc != SQLITE_OK) {
        ERROR("BindSessionTraffParams: bind time failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->time, rc);
        return rc;
    }
    
    rc = sqlite3_bind_int(stmt, idx++, record->handle);
    if (rc != SQLITE_OK) {
        ERROR("BindSessionTraffParams: bind handle failed, idx=%d, value=%d, rc=%d", idx-1, record->handle, rc);
        return rc;
    }
    
    rc = sqlite3_bind_int(stmt, idx++, record->channelId);
    if (rc != SQLITE_OK) {
        ERROR("BindSessionTraffParams: bind channelId failed, idx=%d, value=%u, rc=%d", idx-1, record->channelId, rc);
        return rc;
    }
    
    rc = sqlite3_bind_int(stmt, idx++, record->connId);
    if (rc != SQLITE_OK) {
        ERROR("BindSessionTraffParams: bind connId failed, idx=%d, value=%u, rc=%d", idx-1, record->connId, rc);
        return rc;
    }
    
    rc = sqlite3_bind_int(stmt, idx++, record->sessionId);
    if (rc != SQLITE_OK) {
        ERROR("BindSessionTraffParams: bind sessionId failed, idx=%d, value=%u, rc=%d", idx-1, record->sessionId, rc);
        return rc;
    }
    
    rc = sqlite3_bind_int(stmt, idx++, record->sessionType);
    if (rc != SQLITE_OK) {
        ERROR("BindSessionTraffParams: bind sessionType failed, idx=%d, value=%u, rc=%d", idx-1, record->sessionType, rc);
        return rc;
    }
    
    rc = sqlite3_bind_int(stmt, idx++, record->sendBytes);
    if (rc != SQLITE_OK) {
        ERROR("BindSessionTraffParams: bind sendBytes failed, idx=%d, value=%u, rc=%d", idx-1, record->sendBytes, rc);
        return rc;
    }
    
    rc = sqlite3_bind_int(stmt, idx++, record->recvBytes);
    if (rc != SQLITE_OK) {
        ERROR("BindSessionTraffParams: bind recvBytes failed, idx=%d, value=%u, rc=%d", idx-1, record->recvBytes, rc);
        return rc;
    }
    
    return SQLITE_OK;
}

// TransportTraff绑定参数函数
static int32_t BindTransportTraffParams(sqlite3_stmt *stmt, CommTraffNode *node)
{
    int32_t rc;
    int32_t idx = 1;
    TransportTraff *record = &node->data.data.transportTraff;

    rc = sqlite3_bind_text(stmt, idx++, (const char *)record->ifName,
                           strlen((const char *)record->ifName), SQLITE_STATIC);
    if (rc != SQLITE_OK) {
        ERROR("BindTransportTraffParams: bind ifName failed, idx=%d, value=%s, rc=%d", idx-1, record->ifName, rc);
        return rc;
    }

    rc = sqlite3_bind_text(stmt, idx++, record->local, strlen(record->local), SQLITE_STATIC);
    if (rc != SQLITE_OK) {
        ERROR("BindTransportTraffParams: bind local failed, idx=%d, value=%s, rc=%d", idx-1, record->local, rc);
        return rc;
    }

    rc = sqlite3_bind_text(stmt, idx++, record->remote, strlen(record->remote), SQLITE_STATIC);
    if (rc != SQLITE_OK) {
        ERROR("BindTransportTraffParams: bind remote failed, idx=%d, value=%s, rc=%d", idx-1, record->remote, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->pid);
    if (rc != SQLITE_OK) {
        ERROR("BindTransportTraffParams: bind pid failed, idx=%d, value=%u, rc=%d", idx-1, record->pid, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->time);
    if (rc != SQLITE_OK) {
        ERROR("BindTransportTraffParams: bind time failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->time, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->sockId);
    if (rc != SQLITE_OK) {
        ERROR("BindTransportTraffParams: bind sockId failed, idx=%d, value=%u, rc=%d", idx-1, record->sockId, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->protocol);
    if (rc != SQLITE_OK) {
        ERROR("BindTransportTraffParams: bind protocol failed, idx=%d, value=%u, rc=%d", idx-1, record->protocol, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->sendBytes);
    if (rc != SQLITE_OK) {
        ERROR("BindTransportTraffParams: bind sendBytes failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->sendBytes, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->sendPacks);
    if (rc != SQLITE_OK) {
        ERROR("BindTransportTraffParams: bind sendPacks failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->sendPacks, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->recvBytes);
    if (rc != SQLITE_OK) {
        ERROR("BindTransportTraffParams: bind recvBytes failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->recvBytes, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->recvPacks);
    if (rc != SQLITE_OK) {
        ERROR("BindTransportTraffParams: bind recvPacks failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->recvPacks, rc);
        return rc;
    }

    return SQLITE_OK;
}

// LinkTraff绑定参数函数
static int32_t BindLinkTraffParams(sqlite3_stmt *stmt, CommTraffNode *node)
{
    int32_t rc;
    int32_t idx = 1;
    LinkTraff *record = &node->data.data.linkTraff;

    rc = sqlite3_bind_text(stmt, idx++, (const char *)record->ifName,
                           strlen((const char *)record->ifName), SQLITE_STATIC);
    if (rc != SQLITE_OK) {
        ERROR("BindLinkTraffParams: bind ifName failed, idx=%d, value=%s, rc=%d", idx-1, record->ifName, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->time);
    if (rc != SQLITE_OK) {
        ERROR("BindLinkTraffParams: bind time failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->time, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->txBytes);
    if (rc != SQLITE_OK) {
        ERROR("BindLinkTraffParams: bind txBytes failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->txBytes, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->txPackets);
    if (rc != SQLITE_OK) {
        ERROR("BindLinkTraffParams: bind txPackets failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->txPackets, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->rxBytes);
    if (rc != SQLITE_OK) {
        ERROR("BindLinkTraffParams: bind rxBytes failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->rxBytes, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->rxPackets);
    if (rc != SQLITE_OK) {
        ERROR("BindLinkTraffParams: bind rxPackets failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->rxPackets, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->rxErrPackets);
    if (rc != SQLITE_OK) {
        ERROR("BindLinkTraffParams: bind rxErrPackets failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->rxErrPackets, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->txErrPackets);
    if (rc != SQLITE_OK) {
        ERROR("BindLinkTraffParams: bind txErrPackets failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->txErrPackets, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->rxDropPackets);
    if (rc != SQLITE_OK) {
        ERROR("BindLinkTraffParams: bind rxDropPackets failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->rxDropPackets, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->txDropPackets);
    if (rc != SQLITE_OK) {
        ERROR("BindLinkTraffParams: bind txDropPackets failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->txDropPackets, rc);
        return rc;
    }

    return SQLITE_OK;
}

// RpcTraff绑定参数函数
static int32_t BindRpcTraffParams(sqlite3_stmt *stmt, CommTraffNode *node)
{
    int32_t rc;
    int32_t idx = 1;
    RpcTraff *record = &node->data.data.rpcTraff;

    rc = sqlite3_bind_int(stmt, idx++, record->proxyHandle);
    if (rc != SQLITE_OK) {
        ERROR("BindRpcTraffParams: bind proxyHandle failed, idx=%d, value=%d, rc=%d", idx-1, record->proxyHandle, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->stubHandle);
    if (rc != SQLITE_OK) {
        ERROR("BindRpcTraffParams: bind stubHandle failed, idx=%d, value=%d, rc=%d", idx-1, record->stubHandle, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->sendDataLen);
    if (rc != SQLITE_OK) {
        ERROR("BindRpcTraffParams: bind sendDataLen failed, idx=%d, value=%u, rc=%d", idx-1, record->sendDataLen, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->recvDataLen);
    if (rc != SQLITE_OK) {
        ERROR("BindRpcTraffParams: bind recvDataLen failed, idx=%d, value=%u, rc=%d", idx-1, record->recvDataLen, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->recvDataSeq);
    if (rc != SQLITE_OK) {
        ERROR("BindRpcTraffParams: bind recvDataSeq failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->recvDataSeq, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->sendDataSeq);
    if (rc != SQLITE_OK) {
        ERROR("BindRpcTraffParams: bind sendDataSeq failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->sendDataSeq, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->time);
    if (rc != SQLITE_OK) {
        ERROR("BindRpcTraffParams: bind time failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->time, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->localpid);
    if (rc != SQLITE_OK) {
        ERROR("BindRpcTraffParams: bind localpid failed, idx=%d, value=%d, rc=%d", idx-1, record->localpid, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->localuid);
    if (rc != SQLITE_OK) {
        ERROR("BindRpcTraffParams: bind localuid failed, idx=%d, value=%d, rc=%d", idx-1, record->localuid, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->peerpid);
    if (rc != SQLITE_OK) {
        ERROR("BindRpcTraffParams: bind peerpid failed, idx=%d, value=%d, rc=%d", idx-1, record->peerpid, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->peeruid);
    if (rc != SQLITE_OK) {
        ERROR("BindRpcTraffParams: bind peeruid failed, idx=%d, value=%d, rc=%d", idx-1, record->peeruid, rc);
        return rc;
    }

    return SQLITE_OK;
}

// WpaTraff绑定参数函数
static int32_t BindWpaTraffParams(sqlite3_stmt *stmt, CommTraffNode *node)
{
    int32_t rc;
    int32_t idx = 1;
    WpaTraff *record = &node->data.data.wpaTraff;

    rc = sqlite3_bind_text(stmt, idx++, (const char *)record->ifName,
                           strlen((const char *)record->ifName), SQLITE_STATIC);
    if (rc != SQLITE_OK) {
        ERROR("BindWpaTraffParams: bind ifName failed, idx=%d, value=%s, rc=%d", idx-1, record->ifName, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->signal);
    if (rc != SQLITE_OK) {
        ERROR("BindWpaTraffParams: bind signal failed, idx=%d, value=%d, rc=%d", idx-1, record->signal, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->noise);
    if (rc != SQLITE_OK) {
        ERROR("BindWpaTraffParams: bind noise failed, idx=%d, value=%d, rc=%d", idx-1, record->noise, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->snr);
    if (rc != SQLITE_OK) {
        ERROR("BindWpaTraffParams: bind snr failed, idx=%d, value=%d, rc=%d", idx-1, record->snr, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->txrate);
    if (rc != SQLITE_OK) {
        ERROR("BindWpaTraffParams: bind txrate failed, idx=%d, value=%u, rc=%d", idx-1, record->txrate, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->rxrate);
    if (rc != SQLITE_OK) {
        ERROR("BindWpaTraffParams: bind rxrate failed, idx=%d, value=%u, rc=%d", idx-1, record->rxrate, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->realTxrate);
    if (rc != SQLITE_OK) {
        ERROR("BindWpaTraffParams: bind realTxrate failed, idx=%d, value=%u, rc=%d", idx-1, record->realTxrate, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->realRxrate);
    if (rc != SQLITE_OK) {
        ERROR("BindWpaTraffParams: bind realRxrate failed, idx=%d, value=%u, rc=%d", idx-1, record->realRxrate, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->txEfficiency);
    if (rc != SQLITE_OK) {
        ERROR("BindWpaTraffParams: bind txEfficiency failed, idx=%d, value=%u, rc=%d", idx-1, record->txEfficiency, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->frequency);
    if (rc != SQLITE_OK) {
        ERROR("BindWpaTraffParams: bind frequency failed, idx=%d, value=%u, rc=%d", idx-1, record->frequency, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->txPackets);
    if (rc != SQLITE_OK) {
        ERROR("BindWpaTraffParams: bind txPackets failed, idx=%d, value=%u, rc=%d", idx-1, record->txPackets, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->rxPackets);
    if (rc != SQLITE_OK) {
        ERROR("BindWpaTraffParams: bind rxPackets failed, idx=%d, value=%u, rc=%d", idx-1, record->rxPackets, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->chload);
    if (rc != SQLITE_OK) {
        ERROR("BindWpaTraffParams: bind chload failed, idx=%d, value=%u, rc=%d", idx-1, record->chload, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->txBytes);
    if (rc != SQLITE_OK) {
        ERROR("BindWpaTraffParams: bind txBytes failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->txBytes, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->rxBytes);
    if (rc != SQLITE_OK) {
        ERROR("BindWpaTraffParams: bind rxBytes failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->rxBytes, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->txFailed);
    if (rc != SQLITE_OK) {
        ERROR("BindWpaTraffParams: bind txFailed failed, idx=%d, value=%u, rc=%d", idx-1, record->txFailed, rc);
        return rc;
    }

    rc = sqlite3_bind_int(stmt, idx++, record->txRetries);
    if (rc != SQLITE_OK) {
        ERROR("BindWpaTraffParams: bind txRetries failed, idx=%d, value=%u, rc=%d", idx-1, record->txRetries, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->time);
    if (rc != SQLITE_OK) {
        ERROR("BindWpaTraffParams: bind time failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->time, rc);
        return rc;
    }

    return SQLITE_OK;
}

// Hisysevent绑定参数函数
static int32_t BindHisyseventTraffParams(sqlite3_stmt *stmt, CommTraffNode *node)
{
    int32_t rc;
    int32_t idx = 1;
    HisyseventData *record = &node->data.data.hisyseventData;

    rc = sqlite3_bind_text(stmt, idx++, (const char *)record->domain,
                           strlen((const char *)record->domain), SQLITE_STATIC);
    if (rc != SQLITE_OK) {
        ERROR("BindHisyseventTraffParams: bind domain failed, idx=%d, value=%s, rc=%d", idx-1, record->domain, rc);
        return rc;
    }

    rc = sqlite3_bind_text(stmt, idx++, (const char *)record->event,
                           strlen((const char *)record->event), SQLITE_STATIC);
    if (rc != SQLITE_OK) {
        ERROR("BindHisyseventTraffParams: bind event failed, idx=%d, value=%s, rc=%d", idx-1, record->event, rc);
        return rc;
    }

    rc = sqlite3_bind_text(stmt, idx++, (const char *)record->field,
                           strlen((const char *)record->field), SQLITE_STATIC);
    if (rc != SQLITE_OK) {
        ERROR("BindHisyseventTraffParams: bind field failed, idx=%d, value=%s, rc=%d", idx-1, record->field, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->value);
    if (rc != SQLITE_OK) {
        ERROR("BindHisyseventTraffParams: bind value failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->value, rc);
        return rc;
    }

    rc = sqlite3_bind_int64(stmt, idx++, record->time);
    if (rc != SQLITE_OK) {
        ERROR("BindHisyseventTraffParams: bind time failed, idx=%d, value=%"PRId64", rc=%d", idx-1, record->time, rc);
        return rc;
    }

    return SQLITE_OK;
}

// SessionTraff批量插入函数
int32_t InsertSessionTraffBatch(DbContext *ctx, ListNode *list)
{
    return GenericBatchInsert(ctx, list, SQL_INSERT_SESSION_TRAFF, BindSessionTraffParams, "session");
}

// TransportTraff批量插入函数
int32_t InsertTransportTraffBatch(DbContext *ctx, ListNode *list)
{
    return GenericBatchInsert(ctx, list, SQL_INSERT_TRANSPORT_TRAFF, BindTransportTraffParams, "transport");
}

// LinkTraff批量插入函数
int32_t InsertLinkTraffBatch(DbContext *ctx, ListNode *list)
{
    return GenericBatchInsert(ctx, list, SQL_INSERT_LINK_TRAFF, BindLinkTraffParams, "link");
}

// RpcTraff批量插入函数
int32_t InsertRpcTraffBatch(DbContext *ctx, ListNode *list)
{
    return GenericBatchInsert(ctx, list, SQL_INSERT_RPC_TRAFF, BindRpcTraffParams, "rpc");
}

// WpaTraff批量插入函数
int32_t InsertWpaTraffBatch(DbContext *ctx, ListNode *list)
{
    return GenericBatchInsert(ctx, list, SQL_INSERT_WPA_TRAFF, BindWpaTraffParams, "wpa");
}

// Hisysevent批量插入函数
int32_t InsertHisyseventTraffBatch(DbContext *ctx, ListNode *list)
{
    return GenericBatchInsert(ctx, list, SQL_INSERT_HISYSEVENT_TRAFF, BindHisyseventTraffParams, "hisysevent");
}

// 通用的时间范围查询函数
int32_t QueryTraffByTimeRange(QueryParam *queryParam, CommTraffResult *result)
{
    int32_t rc;
    int32_t num = 0;
    int64_t lastId = 0;

    pthread_mutex_lock(&g_dbMutex);
    if (queryParam == NULL || !CheckDbContextParam(queryParam->ctx) || result == NULL || result->records == NULL) {
        ERROR("invalid parameters");
        pthread_mutex_unlock(&g_dbMutex);
        return COMMTRAFF_INVALID_PARAM;
    }
    char *sql = NULL;
    GetTraffByTimeRangeCb cb = NULL;
    switch(queryParam->traffType) {
        case LINK_LAYER_DATA:
            sql = SQL_SEARCH_LINK_TRAFF_BY_TIMERANGE;
            cb = GetLinkTraffByTimeRangeCb;
            break;
        case TRANSPORT_LAYER_DATA:
            sql = SQL_SEARCH_TRANSPORT_TRAFF_BY_TIMERANGE;
            cb = GetTransportTraffByTimeRangeCb;
            break;
        case SESSION_LAYER_DATA:
            sql = SQL_SEARCH_SESSION_TRAFF_BY_TIMERANGE;
            cb = GetSessionTraffByTimeRangeCb;
            break;
        case RPC_LAYER_DATA:
            sql = SQL_SEARCH_RPC_TRAFF_BY_TIMERANGE;
            cb = GetRpcTraffByTimeRangeCb;
            break;
        case HISYSEVENT_DATA:
            sql = SQL_SEARCH_HISYSEVENT_TRAFF_BY_TIMERANGE;
            cb = GetHisyseventTraffByTimeRangeCb;
            break;
        case WPA_LAYER_DATA:
            sql = SQL_SEARCH_WPA_TRAFF_BY_TIMERANGE;
            cb = GetWpaTraffByTimeRangeCb;
            break;
        default:
            ERROR("invalid traffType");
            pthread_mutex_unlock(&g_dbMutex);
            return COMMTRAFF_INVALID_PARAM;

    }
    rc = QueryData(queryParam->ctx, sql, strlen(sql), BindQueryParamCb, (uint8_t *) queryParam);
    if (rc != SQLITE_ROW) {
        ERROR("QueryData failed,QueryParam startTime=%"PRId64", endTime=%"PRId64" traffType=%d, startRecordIndex=%"PRId64"",
              queryParam->startTime, queryParam->endTime, queryParam->traffType, queryParam->startRecordIndex);
        pthread_mutex_unlock(&g_dbMutex);
        return COMMTRAFF_FAIL;
    }
    //获取数据
    do {
        if (cb(queryParam->ctx, (uint8_t *)result->records, num) != COMMTRAFF_SUCCESS) {
            ERROR("GetSessionTraffByTimeRangeCb failed");
            queryParam->ctx->state &= ~DB_STATE_QUERYING;
            (void)sqlite3_finalize(queryParam->ctx->stmt);
            queryParam->ctx->stmt = NULL;
            pthread_mutex_unlock(&g_dbMutex);
            return COMMTRAFF_FAIL;
        }
        // 获取当前记录的ID，作为lastRecordIndex
        if (sqlite3_column_count(queryParam->ctx->stmt) > 0) {
            lastId = sqlite3_column_int64(queryParam->ctx->stmt, 0);
        }
        rc = QueryDataNext(queryParam->ctx);
        num++;
    } while (rc == SQLITE_ROW && num < ONCE_RECORD_NUMS); // 最多获取ONCE_RECORD_NUMS条记录，与SQL查询中的LIMIT保持一致

    if (rc != SQLITE_DONE && rc != SQLITE_ROW) {
        ERROR("QueryData failed,QueryParam startTime=%"PRId64", endTime=%"PRId64" traffType=%d, startRecordIndex=%"PRId64"",
              queryParam->startTime, queryParam->endTime, queryParam->traffType, queryParam->startRecordIndex);
        pthread_mutex_unlock(&g_dbMutex);
        return COMMTRAFF_FAIL;
    }
    pthread_mutex_unlock(&g_dbMutex);
    result->recordNums = num;
    result->lastRecordIndex = lastId;
    INFO("QueryTraffByTimeRange recordNums=%d, lastRecordIndex=%"PRId64"", num, lastId);
    return COMMTRAFF_SUCCESS;
}
